package com.haizhi.messagehub.service;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import com.haizhi.common.exception.HZException;
import com.haizhi.messagehub.bean.constants.TopicTypeConstants;
import com.haizhi.messagehub.bean.form.ConsumerAdapterForm;
import com.haizhi.messagehub.bean.form.MessageForm;
import com.haizhi.messagehub.service.redis.RedisPubSubProcessor;
import com.haizhi.messagehub.service.redis.RedisQueueProcessor;
import com.haizhi.messagehub.service.redis.RedisStreamProcessor;

@Service
public class MessageHubServiceImpl implements MessageHubService {

    @Resource
    protected StringRedisTemplate stringRedisTemplate;

    @Autowired
    RedisPubSubProcessor redisPubSubProcessor;

    @Autowired
    RedisQueueProcessor redisQueueProcessor;

    @Autowired
    RedisStreamProcessor redisStreamProcessor;

    public Map<String, MessageHubService> messageHubServiceMap = new ConcurrentHashMap<>();


    @PostConstruct
    public void init() {
        messageHubServiceMap.put(TopicTypeConstants.REDIS_PUBSUB_TYPE, redisPubSubProcessor);
        messageHubServiceMap.put(TopicTypeConstants.REDIS_STREAM_TYPE, redisStreamProcessor);
        messageHubServiceMap.put(TopicTypeConstants.REDIS_QUEUE_TYPE, redisQueueProcessor);
    }

    public void checkTopic(String topic, String type) {
        if (!messageHubServiceMap.containsKey(type)) {

            throw new HZException("消息类型不支持");
        }

        List<String> whiteTopicList = stringRedisTemplate.opsForList().range(TopicTypeConstants.WHITE_TOPIC, 0, -1);
        if ((!ObjectUtils.isEmpty(whiteTopicList) && !whiteTopicList.contains(topic)) || ObjectUtils.isEmpty(whiteTopicList)) {
            throw new HZException("当前topic未配置");
        }
    }

    @Override
    public void producer(MessageForm messageForm) {
        this.checkTopic(messageForm.getTopic(), messageForm.getType());
        this.messageHubServiceMap.get(messageForm.getType()).producer(messageForm);
    }


    /**
     * 消费者创建通过注解，已校验topic合法性
     */
    @Override
    public void consumer(ConsumerAdapterForm messageForm) {
        this.messageHubServiceMap.get(messageForm.getType()).consumer(messageForm);
    }
}
